package com.momo.rabbitmq.route.topic;

import com.momo.common.util.snowFlake.SnowFlake;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.core.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 消息路由规则：四种模式（topic、direct、fanout、header）
 * topic：根据绑定关键字通配符规则匹配、比较灵活
 * direct：默认，根据routingKey完全匹配，好处是先匹配再发送
 * fanout：不需要指定routingkey，相当于群发
 * header：不太常用，可以自定义匹配规则
 */

@Component
@Slf4j
public class RabbitTopicSender {

    /**
     * 自动注入RabbitTemplate模板类
     */
    @Autowired
    private RabbitTemplate rabbitTemplate;
    private SnowFlake snowFlake = new SnowFlake(1, 1);

    /**
     * 发送消息方法调用: 构建Message消息
     *
     * @param message
     * @param properties
     * @throws Exception
     */
    public void send(Object messagea, Map<String, Object> properties) {
        MessageHeaders mhs = new MessageHeaders(properties);
        Long id=snowFlake.nextId();
        Message message = MessageBuilder.withBody("".getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("utf-8")
                .setDeliveryTag(id)
                .setMessageId(id.toString()).build();
        rabbitTemplate.setConfirmCallback(confirmCallback);
        rabbitTemplate.setReturnCallback(returnCallback);
        //id + 时间戳 全局唯一
        CorrelationData correlationData = new CorrelationData(id.toString());
        rabbitTemplate.convertAndSend(RabbitMQTopicConfig.EXCHANGE, RabbitMQTopicConfig.ROUTINGKEY, message, correlationData);
    }


    /**
     * 回调函数: confirm确认
     */
    final ConfirmCallback confirmCallback = new ConfirmCallback() {
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            log.info("correlationData:{}; ack:{}; cause:{}", correlationData, ack, cause);
            if (!ack) {
                log.error("correlationData:{}; ack:{}; cause:{}", correlationData, ack, cause);
                log.error("异常处理....");
            }
        }
    };

    /**
     * 回调函数: return返回
     */
    final ReturnCallback returnCallback = new ReturnCallback() {
        @Override
        public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,
                                    String exchange, String routingKey) {
            log.info("return exchange: " + exchange + ", routingKey: "
                    + routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);
        }
    };


}
